package com.we.risk.behavior.repay;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;

import static com.we.risk.behavior.repay.CommonVar.*;

public class TableUser {
    public static final String KFKTABLE = "fdm.uni01_user_risk";
//    public static final String KFKTABLE = "fdm.uni01_user";
    public static final String QUERYSQL =
            "select user_key as account_user_key, "
                    + "ref_id as user_key, "
                    + "update_time "
                    + "from "
                    + KFKTABLE
                    + " where update_time is not null " +
                    "and user_key is not null " +
                    "and ref_id is not null";

    public String account_user_key;
    public String user_key;
    public Timestamp update_time;

    public static Logger LOG = LoggerFactory.getLogger(TableUser.class);

    public TableUser() {
    }

    public static TableUser init(String account_user_key) {
        return new TableUser(account_user_key, null, null);
    }

    public TableUser(String account_user_key, String user_key, Timestamp update_time) {
        this.account_user_key = account_user_key;
        this.user_key = user_key;
        this.update_time = update_time;
    }


    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableUser.QUERYSQL);
        DataStream<Tuple2<Boolean, TableUser>> inputStream =
                tableEnv.toRetractStream(tableRes, TableUser.class);

        SingleOutputStreamOperator<String> out = inputStream.filter(new FilterFunction<Tuple2<Boolean, TableUser>>() {
            @Override
            public boolean filter(Tuple2<Boolean, TableUser> value) throws Exception {
                return value.f0;
            }
        }).uid("TableUser-filter")
                .keyBy(new KeySelector<Tuple2<Boolean, TableUser>, String>() {
                    @Override
                    public String getKey(Tuple2<Boolean, TableUser> value) throws Exception {
                        return value.f1.getAccount_user_key();
                    }
                }).flatMap(new RichFlatMapFunction<Tuple2<Boolean, TableUser>, String>() {
                    ValueState<TableUser> tableUserVState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        tableUserVState = getRuntimeContext()
                                .getState(new ValueStateDescriptor<TableUser>("TableUserVState", TableUser.class));
                    }

                    @Override
                    public void flatMap(Tuple2<Boolean, TableUser> value, Collector<String> out) throws Exception {
                        try {
                            String user_key = value.f1.getUser_key();
                            String account_user_key = value.f1.getAccount_user_key();
                            Timestamp update_time_stamp = value.f1.getUpdate_time();

                            long update_time = update_time_stamp.getTime();

                            TableUser tmpState = tableUserVState.value();
                            TableUser curState = (tmpState == null ? init(account_user_key) : tmpState);

                            Timestamp state_update_time = curState.getUpdate_time();
                            if (state_update_time == null || update_time > state_update_time.getTime()) {
                                String state_user_key = curState.getUser_key();
                                if(state_user_key == null || !state_user_key.equals(user_key)) {
                                    JSONObject resJson = new JSONObject();
                                    resJson.put(ACCUSERKEY, account_user_key);
                                    resJson.put(USERKEY, user_key);

                                    out.collect(resJson.toString());

                                    curState.update_time = update_time_stamp;
                                    curState.user_key = user_key;
                                    tableUserVState.update(curState);
                                }
                            }
                        } catch (Exception e) {
                            LOG.error("TableUser flatMap exec: " + e.toString());
                            LOG.error("TableUser flatMap exec input: " + value.toString());
                        }

                    }
                });

        return out;
    }

    public String getAccount_user_key() {
        return account_user_key;
    }

    public String getUser_key() {
        return user_key;
    }

    public Timestamp getUpdate_time() {
        return update_time;
    }
}
